#include "full_duplex_sticky_bag_echo_server.h"
#include "../log.h"
#include <cstdint>

void Session::Start()
{
    memset(m_read_buf.get(), 0, m_max_len);
    m_socket.async_read_some(boost::asio::buffer(m_read_buf.get(), m_max_len),
    [this, ptr = shared_from_this()](const boost::system::error_code& ec, size_t len){
        _HadnleRead(ec, len, ptr);
    });
}

void Session::Send(char* msg, size_t len)
{
    bool pending = false; // 发送队列是否为空 - false: 当前发送队列为空，可以发送数据
    
    std::unique_lock<std::mutex> lock(m_send_lock);
    if(!m_send_que.empty())
    {
        pending = true;
    }
    if(m_send_que.size() > MAX_SEND_NUM)    //控制发送队列最大长度
    {
        log_error("Session::Send, send queue is full, size: {}", m_send_que.size());
        return;
    }

    m_send_que.push(std::make_shared<MsgNode>(msg, len));

    if(pending) //发送队列不为空，不需要再次调用发送回调函数
    {
        return;
    }

    auto& meg_node = m_send_que.front();
    boost::asio::async_write(m_socket, boost::asio::buffer(meg_node->m_data.get(), meg_node->m_max_len),
        bind(&Session::_HandleWrite, this, std::placeholders::_1, shared_from_this()));
}


void Session::_HadnleRead(const boost::system::error_code& ec, size_t len, std::shared_ptr<Session> self_ptr)
{
    if(ec)
    {
        m_server->DeleteSession(m_uuid);
        log_error("Session::_HadnleRead error, code: {}, message: {}", ec.value(), ec.message());
        return;
    }

    uint32_t copy_len = 0; //已经处理的字节数
    while(len > 0)
    {
        if(!m_head_parse)
        {
            //收到的数据长度小于头部长度
            if(len + m_recv_head_node->m_cur_len < HEAD_LENGTH)
            {
                memcpy(m_recv_head_node->m_data.get() + m_recv_head_node->m_cur_len,
                    m_read_buf.get() + copy_len, len);

                m_recv_head_node->m_cur_len += len;
                memset(m_read_buf.get(), 0, m_max_len);
                
                m_socket.async_read_some(boost::asio::buffer(m_read_buf.get(), m_max_len),
                    [this, ptr = shared_from_this()](const boost::system::error_code& ec, size_t len){
                        _HadnleRead(ec, len, ptr);
                });
                return;
            }

            //收到的数据长度大于头部长度
            uint32_t head_remain = HEAD_LENGTH - m_recv_head_node->m_cur_len; //未处理的头部长度
            memcpy(m_recv_head_node->m_data.get() + m_recv_head_node->m_cur_len,
                m_read_buf.get() + copy_len, head_remain);

            m_recv_head_node->m_cur_len += head_remain;
            copy_len += head_remain;
            len -= head_remain;
            m_head_parse = true;

            //解析头部
            uint32_t date_len = 0;//*(reinterpret_cast<uint32_t*>(m_recv_head_node->m_data.get()));
            ::memcpy(&date_len, m_recv_head_node->m_data.get(), HEAD_LENGTH);
            date_len = boost::asio::detail::socket_ops::network_to_host_long(date_len);
            log_debug("date_len: {}", date_len);
            if(date_len > m_max_len)
            {
                m_server->DeleteSession(m_uuid);
                log_error("data length is too long, close session");
                return;
            }

            m_recv_msg_node = std::make_shared<MsgNode>(date_len);

            //收到的数据长度小于消息体长度
            if(len < date_len)
            {
                memcpy(m_recv_msg_node->m_data.get() + m_recv_msg_node->m_cur_len, 
                    m_read_buf.get() + copy_len, len);
                m_recv_msg_node->m_cur_len += len;
                memset(m_read_buf.get(), 0, m_max_len);
                m_socket.async_read_some(boost::asio::buffer(m_read_buf.get(), m_max_len),
                    [this, ptr = shared_from_this()](const boost::system::error_code& ec, size_t len){
                        _HadnleRead(ec, len, ptr);
                });
                return;
            }

            //收到的数据长度大于等于消息体长度
            memcpy(m_recv_msg_node->m_data.get() + m_recv_msg_node->m_cur_len, 
            m_read_buf.get() + copy_len, date_len);

            m_recv_msg_node->m_cur_len += date_len; 
            copy_len += date_len;
            len -= date_len;
            m_recv_msg_node->m_data[m_recv_msg_node->m_max_len] = '\0';
            log_debug("recv msg: {}", m_recv_msg_node->m_data.get());
            Send(m_recv_msg_node->m_data.get(), m_recv_msg_node->m_max_len);

            //继续处理收到的数据
            m_head_parse = false;
            m_recv_head_node->Clear();
            m_recv_msg_node.reset();
            if(len <= 0)
            {
                memset(m_read_buf.get(), 0, m_max_len);
                m_socket.async_read_some(boost::asio::buffer(m_read_buf.get(), m_max_len),
                    [this, ptr = shared_from_this()](const boost::system::error_code& ec, size_t len){
                        _HadnleRead(ec, len, ptr);
                });
                return;
            }
            continue;
        }

        //处理剩余没有接收完的数据
        //收到的数据依然不足消息体长度
        uint32_t remain_len = m_recv_msg_node->m_max_len - m_recv_msg_node->m_cur_len;
        if(len < remain_len)
        {
            memcpy(m_recv_msg_node->m_data.get() + m_recv_msg_node->m_cur_len,
                m_read_buf.get() + copy_len, len);
            m_recv_head_node->m_cur_len += len;
            memset(m_recv_head_node->m_data.get(), 0, m_max_len);
            m_socket.async_read_some(boost::asio::buffer(m_recv_head_node->m_data.get(), m_max_len),
            [this, ptr = shared_from_this()](const boost::system::error_code& ec, size_t len){
                    _HadnleRead(ec, len, ptr);
            });
            return;
        }

        memcpy(m_recv_msg_node->m_data.get() + m_recv_msg_node->m_cur_len,
            m_read_buf.get() + copy_len, remain_len);
        m_recv_msg_node->m_cur_len += remain_len;
        m_recv_msg_node->m_data[m_recv_msg_node->m_max_len] = '\0';
        copy_len += remain_len;
        len -= remain_len;
        log_info("recv msg: {}", m_recv_msg_node->m_data.get());
        Send(m_recv_msg_node->m_data.get(), m_recv_msg_node->m_max_len);


        m_recv_msg_node.reset();
        m_recv_head_node->Clear();
        m_head_parse = false;
        if(len <= 0)
        {
            memset(m_read_buf.get(), 0, m_max_len);
            m_socket.async_read_some(boost::asio::buffer(m_read_buf.get(), m_max_len),
                [this, ptr = shared_from_this()](const boost::system::error_code& ec, size_t len){
                    _HadnleRead(ec, len, ptr);
            });
            return;
        }
        continue;
    }

}

void Session::_HandleWrite(const boost::system::error_code& ec, std::shared_ptr<Session> self_ptr)
{
    if(ec)
    {
        m_server->DeleteSession(m_uuid);
        log_error("Session::_HandleWrite error, code: {}, message: {}", ec.value(), ec.message());
        return;
    }

    std::unique_lock<std::mutex> lock(m_send_lock);
    m_send_que.pop();
    if(m_send_que.empty())
    {
        return;
    }
    auto msg = m_send_que.front();

    boost::asio::async_write(m_socket, boost::asio::buffer(msg->m_data.get(), msg->m_max_len),
        bind(&Session::_HandleWrite, this, std::placeholders::_1, shared_from_this()));
}

//=============================================================================================================================

Server::Server(boost::asio::io_context&ioc, uint16_t port)
:m_ioc(ioc)
,m_acceptor(ioc, boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port))
{
    _StartAccetp();
    log_debug("server start listening on port: {}", port);
}
void Server::DeleteSession(const std::string& uuid)
{
    std::unique_lock<std::mutex> lock(m_mutex);
    auto it = m_sessions.find(uuid);
    if(it == m_sessions.end())
    {
        return;
    }
    m_sessions.erase(it);
}

//使用智能指针解决程序崩溃问题
void Server::_StartAccetp()
{
    auto session = std::make_shared<Session>(m_ioc, this);
    m_acceptor.async_accept(session->GetSocket(), 
    [this, session](const boost::system::error_code& ec){
        _HandleAccept(session, ec);
    });
}
void Server::_HandleAccept(std::shared_ptr<Session> session, const boost::system::error_code& ec)
{
    if(ec)
    {
        log_error("Server::_HandleAccept error, code: {}, message: {}", ec.value(), ec.message());
        return;
    }

    {
        std::unique_lock<std::mutex> lock(m_mutex);
        m_sessions.insert(std::make_pair(session->GetUuid(), session));
    }
    session->Start();
    _StartAccetp();
}